package demo.java.util.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import demo.java.lang.ThreadDemo;
import demo.java.util.concurrent.ThreadFactoryDemo.NamedThreadFactory;

/**
 * @author hanjy
 *
 */
public class CallableDemo {

    private static Logger logger = LoggerFactory.getLogger(CallableDemo.class);

    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 100, 50, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(), new NamedThreadFactory("xx"));

    private static Random random = new Random();

    public static void main(String[] args) {
        testCallableAndFuture();
    }

    @Test
    public void test1() {
        Future<?> future = threadPool.submit(() -> {
            logger.info("子线程结束");
        });
        ThreadDemo.safeSleep(3000L);
        try {
            future.get(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            logger.error("", e);
        } catch (ExecutionException e) {
            logger.error("", e);
        } catch (TimeoutException e) {
            logger.error("", e);
        }
        logger.info("主线程结束");
    }

    @Test
    public void test并行() {
        List<Future<String>> result = new ArrayList<>();
        IntStream.rangeClosed(0, 9).forEach(e -> {
            Future<String> future = threadPool.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    int n = random.nextInt(5000);
                    logger.info("随机数：{}", n);
                    if (n % 2 == 0) {
                        throw new RuntimeException("偶数异常");
                    }
                    ThreadDemo.safeSleep(n);
                    return "hello " + e;
                }
            });
            result.add(future);

        });

        result.forEach(e -> {
            try {
                logger.info(e.get(1000, TimeUnit.MILLISECONDS));
            } catch (InterruptedException | ExecutionException | TimeoutException e1) {
                logger.error("获取异步结果异常", e1);
            }
        });
    }

    @Test
    public void test串行() {
        List<String> result = new ArrayList<>();
        IntStream.rangeClosed(0, 9).forEach(e -> {
            Future<String> future = threadPool.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    int n = random.nextInt(5000);
                    logger.info("随机数：{}", n);
                    if (n % 2 == 0) {
                        throw new RuntimeException("偶数异常");
                    }
                    ThreadDemo.safeSleep(n);
                    return "hello " + e;
                }
            });

            try {
                result.add(future.get(1000, TimeUnit.MILLISECONDS));
            } catch (InterruptedException | ExecutionException | TimeoutException e1) {
                logger.error("", e1);
            }
        });
        result.forEach(e -> logger.info(e));
    }

    /**
     * Callable接口支持返回执行结果，此时需要调用FutureTask.get()方法实现，此方法会阻塞主线程直到获取‘将来’结果；当不调用此方法时，主线程不会阻塞！
     */
    static void testCallableAndFuture() {

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        Callable<String> callable = new Callable<String>() {
            public String call() throws Exception {

                String threadName = Thread.currentThread().getName();
                Thread.sleep(2000);
                return threadName + " : " + random.nextInt(100);
            }
        };

        Future<String> future = executorService.submit(callable);

        try {
            System.out.println(future.get());
            System.out.println(Thread.currentThread().getName() + " over.");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            System.out.println("会阻塞主线程吗?");
            executorService.shutdown();
        }
    }

}
